package com.uzai.console.config;

import lombok.Data;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MQProducer;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.StringUtils;

/**
 * @author mingyang.chen create in 2023/9/20 13:48
 */

@Configuration
@EnableConfigurationProperties(RocketMQConfig.RocketMQProps.class)
public class RocketMQConfig {

    @Bean
    @ConditionalOnProperty(prefix = "rocketmq2", value = {"name-server", "producer.group"})
    public MQProducer mQProducer2(RocketMQProps rocketMQProperties) {
        RocketMQProperties.Producer producerConfig = rocketMQProperties.getProducer();
        String nameServer = rocketMQProperties.getNameServer();
        String groupName = producerConfig.getGroup();

        String accessChannel = rocketMQProperties.getAccessChannel();

        DefaultMQProducer producer = new DefaultMQProducer();

        producer.setNamesrvAddr(nameServer);
        producer.setUnitName("template2");
        producer.setProducerGroup(groupName);
        if (!StringUtils.isEmpty(accessChannel)) {
            producer.setAccessChannel(AccessChannel.valueOf(accessChannel));
        }
        producer.setSendMsgTimeout(producerConfig.getSendMessageTimeout());
        producer.setRetryTimesWhenSendFailed(producerConfig.getRetryTimesWhenSendFailed());
        producer.setRetryTimesWhenSendAsyncFailed(producerConfig.getRetryTimesWhenSendAsyncFailed());
        producer.setMaxMessageSize(producerConfig.getMaxMessageSize());
        producer.setCompressMsgBodyOverHowmuch(producerConfig.getCompressMessageBodyThreshold());
        producer.setRetryAnotherBrokerWhenNotStoreOK(producerConfig.isRetryNextServer());

        return producer;
    }


    @Bean(destroyMethod = "destroy")
    public RocketMQTemplate rocketMQTemplate2(RocketMQMessageConverter rocketMQMessageConverter, RocketMQProps rocketMQProperties) throws Exception {
        RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
        rocketMQTemplate.setProducer((DefaultMQProducer) mQProducer2(rocketMQProperties));
        rocketMQTemplate.setMessageConverter(rocketMQMessageConverter.getMessageConverter());
        return rocketMQTemplate;
    }

    @ConfigurationProperties(prefix = "rocketmq2")
    @Data
    public static class RocketMQProps {

        /**
         * The name server for rocketMQ, formats: `host:port;host:port`.
         */
        private String nameServer;

        /**
         * Enum type for accessChannel, values: LOCAL, CLOUD
         */
        private String accessChannel;

        private RocketMQProperties.Producer producer;

        private RocketMQProperties.Consumer consumer = new RocketMQProperties.Consumer();
    }
}
